package com.swdegao.iot.netty.mqtt.degaoserver;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.swdegao.iot.netty.utils.ExceptionUitls;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

/**
 * Mqtt订阅端
 */
@Slf4j
@Component
public class MqttPushCallback implements MqttCallback {


    @Autowired
    private MqttConfig mqttConfig;

    private static MqttClient client;

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Value("${spring.mqtt.toRocketMQ}")
    private String toRocketMQ;

    @Value("${spring.mqtt.toRocketMQ_tag}")
    private String toRocketMQ_tag;

    private static MqttClient getClient() {
        return client;
    }

    public static void setClient(MqttClient client) {
        MqttPushCallback.client = client;
    }


    @SneakyThrows
    @Override
    public void connectionLost(Throwable throwable) {
        log.info("连接断开，进行重连");// 连接丢失后，一般在这里面进行重连
        try{
            if (!client.isConnected()) {
                mqttConfig.getMqttPushClient();
                client.reconnect();
            }
        }catch (Exception e){
            log.info("重连异常："+ ExceptionUitls.getExceptionInfo(e));
        }

    }

//    @Override
//    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
//        try {
//            byte[] payload = mqttMessage.getPayload();
//            String mqttMessagePayload = new String(payload);// 接收消息内容(转为字符串格式)
//            // 订阅后得到的消息会执行到这里面
////            log.info("接收消息主题 : " + topic);
//            // log.info("接收消息Qos : " + mqttMessage.getQos());
////            log.info("接收消息内容 : " + mqttMessagePayload);
//            SendResult sr = rocketMQTemplate.syncSend(String.format("%s:%s", toRocketMQ, toRocketMQ_tag), mqttMessagePayload);
////            log.info("mq转发结果 : " + JSONObject.toJSONString(sr));
//        }catch (Exception e){
//            log.error("messageArrived error:{}",e.getMessage());
//            log.error(ExceptionUitls.getExceptionInfo(e));
//        }
//    }


    /**
     * rkmq带key
     * @param topic
     * @param mqttMessage
     * @throws Exception
     */
    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
        try {
            byte[] payload = mqttMessage.getPayload();
            String mqttMessagePayload = new String(payload);// 接收消息内容(转为字符串格式)
            // 订阅后得到的消息会执行到这里面
//            log.info("mq 发布内容 : " + mqttMessagePayload);
            JSONArray res = JSONObject.parseArray(mqttMessagePayload);
            String sn = res.getJSONObject(0).getString("sn");
            Message message = MessageBuilder.withPayload(mqttMessagePayload).setHeader("KEYS", sn).build();
            SendResult sr = rocketMQTemplate.syncSend(String.format("%s:%s", toRocketMQ, toRocketMQ_tag), message);
//            log.info("mq 发布结果 : " + JSONObject.toJSONString(sr));
        }catch (Exception e){
            log.error("messageArrived error:{}",e.getMessage());
            log.error(ExceptionUitls.getExceptionInfo(e));
        }
    }



    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {

        log.info("deliveryComplete---------" + iMqttDeliveryToken.isComplete());
    }

}
